# Copyright (c) 2015 Nicolas JOUANIN
#
# See the file license.txt for copying permission.
import logging
import asyncio
from asyncio import futures
from hbmqtt.mqtt.protocol.handler import ProtocolHandler
from hbmqtt.mqtt.packet import MQTTFixedHeader
from hbmqtt.mqtt.packet import PacketType
from hbmqtt.mqtt.connect import ConnectVariableHeader, ConnectPacket, ConnectPayload
from hbmqtt.mqtt.connack import ConnackPacket
from hbmqtt.mqtt.disconnect import DisconnectPacket
from hbmqtt.mqtt.pingreq import PingReqPacket
from hbmqtt.mqtt.pingresp import PingRespPacket
from hbmqtt.mqtt.subscribe import SubscribePacket
from hbmqtt.mqtt.suback import SubackPacket
from hbmqtt.mqtt.unsubscribe import UnsubscribePacket
from hbmqtt.mqtt.unsuback import UnsubackPacket

class ClientProtocolHandler(ProtocolHandler):
    """
    MQTT 客户端处理
    启动连接，停止连接
    """
    def __init__(self, loop=None):
        super().__init__(loop)
        self._ping_task = None
        self._connack_waiter = None
        self._pingresp_queue = asyncio.Queue()
        self._subscriptions_waiter = dict()
        self._unsubscriptions_waiter = dict()
        self._disconnect_waiter = None
        self._pingresp_waiter = None

    @asyncio.coroutine
    def start(self):
        yield from super().start()
        if self._disconnect_waiter is None:
            self._disconnect_waiter = futures.Future(loop=self._loop)

    @asyncio.coroutine
    def stop(self):
        yield from super().stop()
        if self._ping_task:
            try:
                self._ping_task.cancel()
            except Exception:
                pass
        if self._pingresp_waiter:
            self._pingresp_waiter.cancel()

    def handle_write_timeout(self):
        #>> TODO: 把asyncio.async 当参数看不明白这种写法
        self._ping_task = self._loop.call_soon(asyncio.async, self.mqtt_ping())

    def handle_read_timeout(self):
        pass

    @asyncio.coroutine
    def mqtt_subscribe(self, topics, packet_id):
        """
        :param topics: array of topics [{'filter':'/a/b', 'qos': 0x00}, ...]
        :return:
        """
        #>> 协议打包都独立到单个模块了
        subscribe = SubscribePacket.build(topics, packet_id)
        #>> 将subscribe放到输出队列
        #>> TODO: outgoing_queue在哪里定义的，有什么作用
        yield from self.outgoing_queue.put(subscribe)
        #>> TODO: 这个Future是什么意思, 没有设置回调，没有设置结果
        #>> 理解为发送订阅请求后，创建一个等待任务
        #>> 等待服务器的订阅确认信号, 在 self.handle_suback中处理了
        waiter = futures.Future(loop=self._loop)
        #>> TODO: 添加到字典后又删除，表示不明白
        self._subscriptions_waiter[subscribe.variable_header.packet_id] = waiter
        return_codes = yield from waiter
        del self._subscriptions_waiter[subscribe.variable_header.packet_id]
        return return_codes

    @asyncio.coroutine
    def handle_suback(self, suback: SubackPacket):
        """处理订阅确认包"""
        packet_id = suback.variable_header.packet_id
        try:
            #>> 处理 suback 信号，将结果放到 _subscriptions_waiter中，
            #>> 然后等待 self.mqtt_subscribe 中处理订阅
            #>> 订阅可以在一个客户端中订阅多个主题，因此需要保存
            waiter = self._subscriptions_waiter.get(packet_id)
            waiter.set_result(suback.payload.return_codes)
        except KeyError as ke:
            self.logger.warn("Received SUBACK for unknown pending subscription with Id: %s" % packet_id)

    @asyncio.coroutine
    def mqtt_unsubscribe(self, topics, packet_id):
        """
        取消订阅
        :param topics: array of topics ['/a/b', ...]
        :return:
        """
        #>> 同订阅模式一样，进行打包, 放到输出队列,
        #>> 然后等待确认包
        unsubscribe = UnsubscribePacket.build(topics, packet_id)
        yield from self.outgoing_queue.put(unsubscribe)
        waiter = futures.Future(loop=self._loop)
        self._unsubscriptions_waiter[unsubscribe.variable_header.packet_id] = waiter
        yield from waiter
        del self._unsubscriptions_waiter[unsubscribe.variable_header.packet_id]

    @asyncio.coroutine
    def handle_unsuback(self, unsuback: UnsubackPacket):
        """处理取消订阅的确认包, 同订阅包类似的处理方式"""
        packet_id = unsuback.variable_header.packet_id
        try:
            waiter = self._unsubscriptions_waiter.get(packet_id)
            waiter.set_result(None)
        except KeyError as ke:
            self.logger.warn("Received UNSUBACK for unknown pending subscription with Id: %s" % packet_id)

    @asyncio.coroutine
    def mqtt_connect(self):
        def build_connect_packet(session):
            """构建连接的包"""
            #>> TODO: 这个方法应该放到协议包中
            vh = ConnectVariableHeader()
            payload = ConnectPayload()

            vh.keep_alive = session.keep_alive
            vh.clean_session_flag = session.clean_session
            vh.will_retain_flag = session.will_retain
            payload.client_id = session.client_id

            if session.username:
                vh.username_flag = True
                payload.username = session.username
            else:
                vh.username_flag = False

            if session.password:
                vh.password_flag = True
                payload.password = session.password
            else:
                vh.password_flag = False
            if session.will_flag:
                vh.will_flag = True
                vh.will_qos = session.will_qos
                payload.will_message = session.will_message
                payload.will_topic = session.will_topic
            else:
                vh.will_flag = False

            header = MQTTFixedHeader(PacketType.CONNECT, 0x00)
            packet = ConnectPacket(header, vh, payload)
            return packet

        packet = build_connect_packet(self.session)
        yield from self.outgoing_queue.put(packet)
        self._connack_waiter = futures.Future(loop=self._loop)
        return (yield from self._connack_waiter)

    @asyncio.coroutine
    def handle_connack(self, connack: ConnackPacket):
        #>> 一个客户端，只有一个连接，因此不需要像处理订阅的那种方式保存waiter
        #>> 直接回应确认包即可
        self._connack_waiter.set_result(connack.variable_header.return_code)

    @asyncio.coroutine
    def mqtt_disconnect(self):
        """同连接一样的处理断开连接"""
        # yield from self.outgoing_queue.join() To be used in Python 3.5
        disconnect_packet = DisconnectPacket()
        yield from self.outgoing_queue.put(disconnect_packet)
        self._connack_waiter = None

    @asyncio.coroutine
    def mqtt_ping(self):
        """心跳请求"""
        ping_packet = PingReqPacket()
        yield from self.outgoing_queue.put(ping_packet)
        self._pingresp_waiter = futures.Future(loop=self._loop)
        resp = yield from self._pingresp_queue.get()
        self._pingresp_waiter = None
        return resp

    @asyncio.coroutine
    def handle_pingresp(self, pingresp: PingRespPacket):
        """心跳响应"""
        yield from self._pingresp_queue.put(pingresp)

    @asyncio.coroutine
    def handle_connection_closed(self):
        self.logger.debug("Broker closed connection")
        if not self._disconnect_waiter.done():
            self._disconnect_waiter.set_result(None)

    @asyncio.coroutine
    def wait_disconnect(self):
        yield from self._disconnect_waiter
